package org.zhaya.sdk.service;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.exceptions.ExceptionUtil;
import com.google.common.collect.Lists;
import io.logz.sawmill.Doc;
import io.logz.sawmill.ExecutionResult;
import io.logz.sawmill.PipelineExecutionTimeWatchdog;
import io.logz.sawmill.exceptions.PipelineExecutionException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.zhaya.sdk.ProcessorInterceptor;
import org.zhaya.sdk.constant.ResultEnum;
import org.zhaya.sdk.executor.PipelineExecutor;
import org.zhaya.sdk.executor.PipelineResult;
import org.zhaya.sdk.model.ExecuteResultDetail;
import org.zhaya.sdk.model.PipelineExecuteResult;
import org.zhaya.sdk.model.PipelineNode;
import org.zhaya.sdk.processors.DebugProcessor;
import org.zhaya.sdk.processors.TransposeProcessor;
import org.zhaya.sdk.swamill.CustomPipeline;
import org.zhaya.sdk.swamill.NullPipelineExecutionMetricsTracker;
import org.zhaya.sdk.swamill.ProcessDoc;
import org.zhaya.sdk.util.ClonerUtils;
import org.zhaya.sdk.util.PipelineUtils;

import java.util.List;
import java.util.Map;
import java.util.ServiceLoader;
import java.util.stream.Collectors;

/**
 * @Author Ryan Yang
 * @Date 2022/10/10 4:25 下午
 */
@Slf4j
public class PipelineExecuteService {

    private static final PipelineExecutor pipelineExecutor;

    static {
        List<ProcessorInterceptor> processorInterceptors = Lists.newArrayList();
        ServiceLoader<ProcessorInterceptor> interceptors = ServiceLoader.load(ProcessorInterceptor.class);
        for (ProcessorInterceptor interceptor : interceptors) {
            processorInterceptors.add(interceptor);
        }
        pipelineExecutor = new PipelineExecutor(new PipelineExecutionTimeWatchdog(100, 3000, new NullPipelineExecutionMetricsTracker(), context -> {
        }), new NullPipelineExecutionMetricsTracker(), processorInterceptors);
    }

    /**
     * 模型执行
     *
     * @param modelJson 模型json
     * @param rawData   原始数据（单条）
     * @return 执行结果
     */
    public static PipelineExecuteResult execute(String modelJson, String rawData) {
        List<PipelineNode> pipelineNodeList = PipelineParseService.parse(modelJson);
        PipelineParseService.parse(pipelineNodeList);
        String pipelineConf = PipelineParseService.convertToConfig(pipelineNodeList);
        CustomPipeline customPipeline = PipelineUtils.getPipeline(null, pipelineConf);
        ProcessDoc doc = PipelineUtils.createDocForMessage(rawData);
        PipelineExecuteResult executeResult = new PipelineExecuteResult();
        long time = System.nanoTime();
        PipelineResult result = pipelineExecutor.execute(customPipeline, doc);
        ExecutionResult executionResult = result.getResult();
        executeResult.setTime(DateUtil.spendNt(time));
        executeResult.setPipelineExecutionResult(buildPipelineResultDetail(executionResult));
        if (CollUtil.isNotEmpty(result.getDocSet())) {
            List<Map<String, Object>> docSourceList = result.getDocSet().stream().map(Doc::getSource).collect(Collectors.toList());
            executeResult.setData(docSourceList.size() == 1 ? docSourceList.get(0) : docSourceList);
        }
        return executeResult;
    }

    /**
     * 调试执行
     */
    public static List<PipelineExecuteResult> executeOnce(String pipelineConf, List<ProcessDoc> docs) {
        List<PipelineExecuteResult> resultList = Lists.newArrayListWithExpectedSize(docs.size());
        try {
            CustomPipeline customPipeline = PipelineUtils.getPipeline(null, pipelineConf);
            for (ProcessDoc doc : docs) {
                PipelineExecuteResult es = new PipelineExecuteResult();
                long time = System.nanoTime();
                ExecutionResult executionResult = pipelineExecutor.execute(customPipeline, doc).getResult();
                es.setTime(DateUtil.spendNt(time));
                es.setPipelineExecutionResult(buildPipelineResultDetail(executionResult));
                resultList.add(es);
                try {
                    List<Doc> debugResult = ClonerUtils.deepClone(DebugProcessor.threadLocal.get());
                    log.debug("Debug Result : {}", debugResult);
                    if (CollUtil.isEmpty(debugResult)) {
                        continue;
                    }
                    if (CollUtil.size(debugResult) > 1) {
                        es.setData(debugResult.stream().map(Doc::getSource).collect(Collectors.toList()));
                    } else {
                        Doc cloneDoc = debugResult.get(0);
                        Map<String, Object> source = cloneDoc.getSource();
                        if (source.containsKey(TransposeProcessor.MULTI_DOC_FIELD)) {
                            Object docsSource = source.get(TransposeProcessor.MULTI_DOC_FIELD);
                            if (CollUtil.size(docsSource) > 1) {
                                Object docsSourceList = ((List) docsSource).stream().map(debug -> ((Doc) debug).getSource()).collect(Collectors.toList());
                                es.setData(docsSourceList);
                            } else {
                                Doc clone2Doc = (Doc) (((List) docsSource).get(0));
                                es.setData(clone2Doc.getSource());
                            }
                        } else {
                            es.setData(cloneDoc.getSource());
                        }
                    }
                } catch (Exception e) {
                    log.error("Debug Process Failed:", e);
                } finally {
                    DebugProcessor.threadLocal.remove();
                }
            }
        } catch (Exception e) {
            throw ExceptionUtil.wrapRuntime(StringUtils.defaultString(e.getMessage()));
        }
        return resultList;
    }

    public static ExecuteResultDetail buildPipelineResultDetail(ExecutionResult executionResult) {
        ExecuteResultDetail resultDetail = new ExecuteResultDetail();

        java.util.Optional<ExecutionResult.Error> errorOptional = executionResult.getError();
        if (errorOptional.isPresent()) {
            ExecutionResult.Error error = errorOptional.get();
            java.util.Optional<PipelineExecutionException> executionException = error.getException();
            if (executionException.isPresent()) {
                PipelineExecutionException pipelineExecutionException = executionException.get();
                resultDetail.setException(ExceptionUtil.getMessage(pipelineExecutionException));
            }
            resultDetail.setFailedProcessorName(error.getFailedProcessorName());
            resultDetail.setErrorMessage(error.getMessage());
        }

        java.util.Optional<Long> overtimeTookOptional = executionResult.getOvertimeTook();
        resultDetail.setOvertimeTook(overtimeTookOptional.orElse(null));

        ResultEnum resultEnum;
        if (executionResult.isSucceeded()) {
            resultEnum = ResultEnum.SUCCEEDED;
        } else if (executionResult.isDropped()) {
            resultEnum = ResultEnum.DROPPED;
        } else if (executionResult.isExpired()) {
            resultEnum = ResultEnum.EXPIRED;
        } else {
            resultEnum = ResultEnum.FAILED;
        }
        resultDetail.setResult(resultEnum.name());
        return resultDetail;
    }
}
